package org.example.first_demo.kafka_flink;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

import java.util.Properties;

public class Flink {

    public static void main(String[] args) throws Exception {

        // 创建 Flink 流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 配置 Kafka
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("group.id", "test-group");

        // 创建 Kafka Consumer
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
                "test-topic", // Kafka topic
                new SimpleStringSchema(), // 消费的消息类型
                properties
        );

        // 从 Kafka 读取数据
        DataStream<String> stream = env.addSource(consumer);

        // 数据转换（简单的打印）
        stream.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) {
                // 这里只是简单的转换，可以进行更复杂的操作
                return "Processed: " + value;
            }
        }).print();

        // 执行 Flink 程序
        env.execute("Flink Kafka Example");
    }
}
